package cn.edu.cuit.elena.hdfs;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.Reader;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;

import com.google.gson.Gson;

import cn.edu.cuit.elena.common.Identifiable;
import cn.edu.cuit.elena.common.SackConstants;
import cn.edu.cuit.elena.trace.Tracer;

public class HDFSManager
{
    public static final String HDFS_ROOT = "hdfs://localhost:8020";
    public static final String HIVE_ROOT = SackConstants.HDFS_HIVE_TABLES_ROOT_DIR;

    private static HDFSManager hdfsManager;

    private FileSystem hdfs;

    private HDFSManager() throws IOException, URISyntaxException
    {
        Configuration configuration = new Configuration();
        hdfs = FileSystem.get( new URI( "hdfs://localhost:8020/" ), configuration );
    }

    public static HDFSManager instance()
    {
        try
        {
            if( hdfsManager == null )
            {
                hdfsManager = new HDFSManager();
            }

            return hdfsManager;
        }
        catch( Exception e )
        {
            Tracer.fatalTrace( HDFSManager.class, "HDFSManage can't get hdfs due to " + e.getMessage(), e );
            return null;
        }
    }

    /**
     * Verify a specified path begin with "/..." exited or not, the given path will be spelled to a
     * form like this : "hdfs://localhost:8020/user/quintus/...". If it's a qualified path, and
     * nothing will do.
     * */
    public boolean exsits( String path )
    {
        path = toQualifiedPath( path );

        try
        {
            return hdfs.exists( new Path( path ) );
        }
        catch( Exception e )
        {
            Tracer.fatalTrace( getClass(), "HDFSManager path check fail due to : " + e.getMessage(), e );
            return false;
        }
    }

    public FileSystem hdfs()
    {
        return hdfs;
    }

    /**
     * Get path for each file in the given directory path, if there is any sub-directory, it would
     * be ignored.
     * 
     * @param dirPath
     *            Directory path should begin with "/", and it should be the path related to the
     *            HDFS_ROOT "hdfs://localhost"
     * */
    public String[] getAllFiles( String dirPath )
    {

        FileStatus[] fileStatuses = null;
        try
        {
            fileStatuses = hdfs.listStatus( new Path( dirPath ) );
        }
        catch( Exception e )
        {
            Tracer
                .fatalTrace( HDFSManager.class, "HDFSManager exception ignored : hdfs exception " + e.getMessage(), e );
        }

        if( fileStatuses == null )
        {

            throw new NullPointerException( "hdfs exception" );
        }
        Path[] pathes = FileUtil.stat2Paths( fileStatuses );

        LinkedList<String> temp = new LinkedList<String>();

        for( Path path : pathes )
        {
            try
            {
                if( hdfs.isFile( path ) )
                {
                    temp.add( path.toString() );
                }
            }
            catch( IOException e )
            {
                // There should not be any exceptions.
                Tracer.fatalTrace( HDFSManager.class,
                    "HDFSManager exception ignored : hdfs exception " + e.getMessage(), e );
            }
        }

        String[] tempArray = new String[temp.size()];

        for( int i = 0; i < temp.size(); i++ )
        {
            tempArray[i] = temp.get( i );
        }
        return tempArray;
    }

    public String[] getAllSubDirectory( String dirPath )
    {
        FileStatus[] fileStatuses = null;
        try
        {
            fileStatuses = hdfs.listStatus( new Path( dirPath ) );
        }
        catch( Exception e )
        {
            Tracer
                .fatalTrace( HDFSManager.class, "HDFSManager exception ignored : hdfs exception " + e.getMessage(), e );
        }

        if( fileStatuses == null )
        {

            throw new NullPointerException( "hdfs exception" );
        }
        Path[] pathes = FileUtil.stat2Paths( fileStatuses );

        LinkedList<String> temp = new LinkedList<String>();

        for( Path path : pathes )
        {
            try
            {
                if( hdfs.isFile( path ) )
                {
                    continue;
                }
                temp.add( path.toString() );
            }
            catch( IOException e )
            {
                // There should not be any exceptions.
                Tracer.fatalTrace( HDFSManager.class,
                    "HDFSManager exception ignored : hdfs exception " + e.getMessage(), e );
            }
        }

        String[] tempArray = new String[temp.size()];

        for( int i = 0; i < temp.size(); i++ )
        {
            tempArray[i] = temp.get( i );
        }
        return tempArray;
    }

    public String[] getAll( String dirPath )
    {
        FileStatus[] fileStatuses = null;
        try
        {
            fileStatuses = hdfs.listStatus( new Path( dirPath ) );
        }
        catch( Exception e )
        {
            Tracer
                .fatalTrace( HDFSManager.class, "HDFSManager exception ignored : hdfs exception " + e.getMessage(), e );
        }

        if( fileStatuses == null )
        {

            throw new NullPointerException( "hdfs exception" );
        }
        Path[] pathes = FileUtil.stat2Paths( fileStatuses );

        LinkedList<String> temp = new LinkedList<String>();

        for( Path path : pathes )
        {
            temp.add( path.toString() );
        }

        String[] tempArray = new String[temp.size()];

        for( int i = 0; i < temp.size(); i++ )
        {
            tempArray[i] = temp.get( i );
        }
        return tempArray;
    }

    public static String toQualifiedPath( String path )
    {
        path = path.trim();
        if( path.startsWith( HDFS_ROOT ) )
        {
            return path;
        }

        if( path.startsWith( HIVE_ROOT ) )
        {
            return HDFS_ROOT + path;
        }

        return HDFS_ROOT + HIVE_ROOT + path;
    }

    public void create( String path ) throws IOException
    {
        hdfs.create( new Path( path ) );
    }

    public void delete( String path ) throws IOException
    {
        hdfs.delete( new Path( path ), true );
    }

    public void moveFromLocalFile( String src, String tar ) throws IOException
    {
        hdfs.moveFromLocalFile( new Path( src ), new Path( tar ) );
    }

    public void moveToLocal( String src, String tar ) throws IOException
    {
        hdfs.moveToLocalFile( new Path( src ), new Path( tar ) );
    }

    public void copyToLocalFile( String src, String tar ) throws IOException
    {
        hdfs.copyToLocalFile( new Path( src ), new Path( tar ) );
    }

    public void copyFromLocalFile( String src, String tar ) throws IOException
    {
        hdfs.copyFromLocalFile( new Path( src ), new Path( tar ) );
    }

    public boolean legalToAccess( Identifiable id, String path )
    {
        //TODO
        return false;
    }

    public void chown( Identifiable id, String path )
    {
        //TODO
    }

    public Object getDirTree( String dirPath )
    {
        if( getAllSubDirectory( dirPath ).length == 0 )
        {
            List<Map<String, String>> list = new LinkedList<Map<String, String>>();
            String[] files = getAllFiles( dirPath );
            for( String file : files )
            {
                HashMap<String, String> map = new HashMap<String, String>();
                map.put( "text", file );
                map.put( "leaf", "true" );
                list.add( map );
            }

            return list;
        }

        String[] subDirs = getAllSubDirectory( dirPath );

        List<Map<String, ? extends Object>> list = new LinkedList<Map<String, ? extends Object>>();
        for( String subDir : subDirs )
        {
            HashMap<String, Object> map = new HashMap<String, Object>();
            map.put( "text", subDir );
            map.put( "children", getDirTree( subDir ) );
        }

        String[] currentFiles = getAllFiles( dirPath );
        for( String file : currentFiles )
        {
            Map<String, String> map = new HashMap<String, String>();
            map.put( "text", file );
            map.put( "leaf", "true" );
            list.add( map );
        }
        return list;
    }

    public List<Map<String, String>> getHDFSData( String path ) throws IOException
    {
        FSDataInputStream fsDataInputStream = hdfs.open( new Path( path ) );
        InputStreamReader inputStreamReader = new InputStreamReader( fsDataInputStream );
        BufferedReader bufferedReader = new BufferedReader( inputStreamReader );

        List<Map<String, String>> list = new LinkedList<Map<String, String>>();

        String line = null;
        while( ( line = bufferedReader.readLine() ) != null )
        {
            int column = 0;
            Map<String, String> map = new HashMap<String, String>();
            String[] ss = line.split( "'" );
            for( String s : ss )
            {
                column++;
                map.put( String.valueOf( column ), s );
            }
            list.add( map );
        }

        return list;
    }

    public List<Map<String, String>> getHDFSData( String path, int count ) throws IOException
    {
        FSDataInputStream fsDataInputStream = hdfs.open( new Path( path ) );
        InputStreamReader inputStreamReader = new InputStreamReader( fsDataInputStream );
        BufferedReader bufferedReader = new BufferedReader( inputStreamReader );

        List<Map<String, String>> list = new LinkedList<Map<String, String>>();

        String line = null;
        int init = 0;
        while( ( line = bufferedReader.readLine() ) != null && init <= count )
        {
            int column = 0;
            Map<String, String> map = new HashMap<String, String>();
            String[] ss = line.split( "'" );
            for( String s : ss )
            {
                column++;
                map.put( String.valueOf( column ), s );
            }
            list.add( map );
            init++;
        }

        return list;
    }
}
